package com.atguigu.flink.sql.query;

import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Over;
import org.apache.flink.table.api.OverWindow;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.*;

/**
 * Created by 黄凯 on 2023/6/26 0026 20:58
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * Flink SQL的窗口:
 * *    1. 分组窗口
 * *         滚动窗口( 时间 、计数)
 * *         滑动窗口( 时间 、计数)
 * *         会话窗口( 时间)
 * *
 * *         定义窗口语法:
 * *            .window(Tumble.over(lit(10).minutes()).on($("rowtime")).as("w"));
 * *         使用窗口语法:
 * *            Table table = input
 * *              .window([GroupWindow w].as("w"))  // 定义窗口并指定别名为 w
 * *              .groupBy($("w"))  // 以窗口 w 对表进行分组
 * *              .select($("b").sum());  // 聚合
 * *
 * *
 * *    2. Window TVF(窗口表值函数)
 * *         滚动窗口
 * *         滑动窗口
 * *         累积窗口
 * *         会话窗口(暂不支持)
 * *
 * *    3. Over(开窗)
 * *         TableAPI:
 * *            Over.partitionBy($("a")).orderBy($("proctime")).preceding(UNBOUNDED_ROW).as("w")
 * *         SQL:
 * *            SELECT
 * *              agg_func(agg_col) OVER (
 * *                    [PARTITION BY col1[, col2, ...]]
 * *                    ORDER BY time_col
 * *                    range_definition),
 * *              ...
 * *            FROM .
 */
public class Flink05_OverWindowTableAPI {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> ds = env.socketTextStream("127.0.0.1", 8888)
                .map(
                        line -> {
                            String[] fields = line.split(",");
                            return new WaterSensor(fields[0].trim(), Long.valueOf(fields[1].trim()), Integer.valueOf(fields[2].trim()));
                        }

                );
        //创建表环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //流转表
        Schema schema = Schema.newBuilder()
                .column("id", "STRING")
                .column("vc", "INT")
                .column("ts", "BIGINT")
                .columnByExpression("pt", "PROCTIME()")
                .columnByExpression("et", "TO_TIMESTAMP_LTZ(ts,3)")
                .watermark("et", "et - INTERVAL '0' SECOND")
                .build();
        Table table = tableEnv.fromDataStream(ds, schema);
        table.printSchema();

        // 基于行，上无边界下无边界(错误的)
        //OVER RANGE FOLLOWING windows are not supported yet.
        OverWindow w1 =
                Over.partitionBy($("id")).orderBy($("pt")).preceding(UNBOUNDED_ROW).following(UNBOUNDED_ROW).as("w");
        // 基于行，上无边界到当前行
        OverWindow w2 =
                Over.partitionBy($("id")).orderBy($("pt")).preceding(UNBOUNDED_ROW).following(CURRENT_ROW).as("w");
        // 基于行，上有边界到当前行
        OverWindow w3 =
                Over.partitionBy($("id")).orderBy($("pt")).preceding(rowInterval(2L)).following(CURRENT_ROW).as("w");


        //基于时间
        //上无边界， 下午边界(错误）
        // OVER RANGE FOLLOWING windows are not supported yet.
        OverWindow w4 =
                Over.partitionBy($("id")).orderBy($("et")).preceding(UNBOUNDED_RANGE).following(UNBOUNDED_RANGE).as("w");
        //上无边界到当前时间
        OverWindow w5 =
                Over.partitionBy($("id")).orderBy($("et")).preceding(UNBOUNDED_RANGE).following(CURRENT_RANGE).as("w");
        //上有边界到当前时间
        OverWindow w6 =
                Over.partitionBy($("id")).orderBy($("et")).preceding(lit(2).seconds()).following(CURRENT_RANGE).as("w");

        //使用窗口
        table.window(w6)
                .select( $("id"), $("vc") , $("ts") ,  $("id").count().over($("w")).as("cnt") )
                .execute()
                .print();

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }

}
